package com.alibaba.otter.node.etl.load.loader.mq.rabbit;

import com.alibaba.otter.node.etl.load.loader.mq.rabbit.api.ChannelManager;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.assembly.MessageAssembler;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.callback.*;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.exception.MqPluginException;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.model.Message;
import com.google.common.collect.Lists;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ShutdownSignalException;
import com.rabbitmq.client.impl.AMQImpl;
import lombok.AccessLevel;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.List;

@Slf4j
public class RabbitMqPublisher implements Publisher {

    private ChannelManager channelManager = new ChannelManager();

    @Setter(AccessLevel.PACKAGE)
    private PublishFailureCallback failureCallback = new DefaultFailureCallback();

    @Setter(AccessLevel.PACKAGE)
    private boolean requireConfirm = true;

    @Setter(AccessLevel.PACKAGE)
    @Getter
    private String exchange;

    @Setter(AccessLevel.PACKAGE)
    @Getter
    private String routingKey = "";

    @Setter(AccessLevel.PACKAGE)
    private String namespaceName;

    /**
     * 当mandatory标志位设置为true时，如果exchange根据自身类型和消息routeKey无法找到一个符合条件的queue， 那么会调用basic.return方法将消息返回给生产者<br>
     * 当mandatory设置为false时，出现上述情形broker会直接将消息扔掉。
     */
    @Setter(AccessLevel.PACKAGE)
    private boolean mandatory = true;

    /**
     * 当immediate标志位设置为true时，如果exchange在将消息路由到queue(s)时发现对于的queue上么有消费者， 那么这条消息不会放入队列中。当与消息routeKey关联的所有queue（一个或者多个）都没有消费者时，
     * 该消息会通过basic.return方法返还给生产者。
     */
    @Setter(AccessLevel.PACKAGE)
    private boolean immediate;

    @Setter
    private MessageAssembler assembler;

    public RabbitMqPublisher(String namespaceName, MessageAssembler assembler) {
        this.namespaceName = namespaceName;
        this.assembler = assembler;
    }

    @Override
    public Object send(Object message) {
        if (message == null) {
            log.warn("消息不能为null");
            return null;
        }
        return doPublish(assembler.assemblyMessage(message), message);
    }

    private Object doPublish(Message message, Object rawMsg) {

        Channel channel = channelManager.getPublisherChannel(namespaceName);
        ProxiedConfirmListener confirmListener = new ProxiedConfirmListener();
        confirmListener.setChannelManager(channelManager);
        confirmListener.setChannel(channel);
        confirmListener.setNamespace(namespaceName);
        confirmListener.addSuccessCallbacks(successCallbacks);
        channel.addConfirmListener(confirmListener);

        AMQP.BasicProperties messageProperties = null;
        if (message.getProperty() instanceof AMQP.BasicProperties) {
            messageProperties = (AMQP.BasicProperties) message.getProperty();
        }
        long nextSeq = -1;
        try {
            if (requireConfirm) {
                channel.confirmSelect();
            }
            nextSeq = channel.getNextPublishSeqNo();
            if (requireConfirm) {
                confirmListener.toConfirm(nextSeq, rawMsg);
            }
            channel.basicPublish(
                exchange, routingKey,
                mandatory, immediate,
                messageProperties,
                message.getContent()
            );
            if (!requireConfirm) {
                channelManager.releaseChannel(channel, namespaceName);
            }
        } catch (ShutdownSignalException e) {
            // 如果因为网络问题或远端关闭，导致发送失败。则进行重连和重发。
            confirmListener.remove(nextSeq);
            releaseChannel(channel);
            handleShutdownSignal(message, e);
            MqUtils.sleep(MqConstants.RESEND_INTERVAL_IN_MS);
            doPublish(message, rawMsg);
        } catch (IOException e) {
            confirmListener.remove(nextSeq);
            releaseChannel(channel);
            if (e.getCause() instanceof ShutdownSignalException) {
                ShutdownSignalException ex = (ShutdownSignalException) e.getCause();
                handleShutdownSignal(message, ex);
            }
            log.error("发送消息失败", e);
            MqUtils.sleep(MqConstants.RESEND_INTERVAL_IN_MS);
            doPublish(message, rawMsg);
        } catch (Throwable e) {
            confirmListener.remove(nextSeq);
            releaseChannel(channel);
            failureCallback.onUnexpectedError(e);
        }
        // 无返回值，只为与原接口保持兼容。
        return null;
    }

    private void releaseChannel(Channel channel) {
        channelManager.releaseChannel(channel, namespaceName);
    }

    private void handleShutdownSignal(Message message, ShutdownSignalException ex) {
        if (ex.getReason() instanceof AMQImpl.Channel.Close) {
            AMQImpl.Channel.Close reason = (AMQImpl.Channel.Close) ex.getReason();
            handleShutdownErrorCode(message, ex, reason);
        }
    }

    private void handleShutdownErrorCode(Message message, ShutdownSignalException ex, AMQImpl.Channel.Close reason) {
        if (reason.getReplyCode() == MqConstants.SHUTDOWN_SIGNAL_NOT_FOUND) {
            failureCallback.onNoDestination(exchange, routingKey, message);
            throw new MqPluginException("没有exchange或routingKey", ex);
        }
    }

    private List<SuccessCallback> successCallbacks = Lists.newArrayList();

    public void addSuccessCallbacks(List<SuccessCallback> successCallbacks) {
        this.successCallbacks.addAll(successCallbacks);
    }

    public void setConnectedCallback(ConnectedCallback connectedCallback) {
        channelManager.setConnectedCallback(connectedCallback);
    }

    public void setDisconnectedCallback(DisconnectedCallback disconnectedCallback) {
        channelManager.setDisconnectedCallback(disconnectedCallback);
    }
}
